#pragma once
#include "../common/net.hpp"
#include "../common/message.hpp"
#include <unordered_set>

namespace RPC
{
    namespace server
    {
        class TopicManager
        {
        public:
            using ptr = std::shared_ptr<TopicManager>;
            TopicManager() {};
            void onTpoicRequest(const BaseConnection::ptr &conn, TopicRequest::ptr &msg)
            {
                // 主题创建;
                TopicOpType topic_optype = msg->Optype();
                bool ret = true;
                switch(topic_optype)
                {
                    case TopicOpType::TOPIC_CREATE: topicCreate(conn, msg); break;
                    case TopicOpType::TOPIC_REMOVE: topicRemove(conn, msg); break;
                    case TopicOpType::TOPIC_SUBSCRIBE: ret = topicSubscribe(conn, msg); break;
                    case TopicOpType::TOPIC_CANCEL: topicCancel(conn, msg); break;
                    case TopicOpType::TOPIC_PUBLISH: ret = topicPublish(conn, msg); break;
                    default:   return errorResponse(conn, msg, RCode::RCODE_INVALID_OPTYPE);
                }
                if(!ret) return errorResponse(conn, msg, RCode::RCODE_NOT_FOUND_TOPIC);
                return topicResponse(conn, msg);
            }

            // 一个订阅者
            void onShutdown(const BaseConnection::ptr &conn)
            {
                //消息发布者断开连接不需要操作, 消息订阅者删除需要隔离;
                std::vector<Topic::ptr> topics;
                Subscriber::ptr subscriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _subscribes.find(conn);
                    if(it == _subscribes.end())
                    {
                        return;
                    }

                    subscriber = it->second;
                    for(auto& topic_name : it->second->_topics)
                    {
                        auto topic_it = _topics.find(topic_name);
                        if(topic_it == _topics.end()) continue;
                        topics.push_back(topic_it->second);
                    }
                    //订阅者里面删除;
                    _subscribes.erase(it);
                }
                //从受影响的主题对象中删除订阅者;
                for(auto& topic : topics)
                {
                    topic->removeSubscriber(subscriber);
                }
            }

        private:
            void errorResponse(const BaseConnection::ptr& conn, const TopicRequest::ptr& msg, RCode rcode)
            {
                auto msg_rsp = MessageFactory::create<TopicResponse>();
                msg_rsp->setId(msg->rid());
                msg_rsp->setMType(MType::RSP_TOPIC);
                msg_rsp->setRCode(rcode);
                conn->send(msg_rsp);
            }

            void topicResponse(const BaseConnection::ptr& conn, const TopicRequest::ptr& msg)
            {
                auto msg_rsp = MessageFactory::create<ServiceResponse>();
                msg_rsp->setId(msg->rid());
                msg_rsp->setMType(MType::RSP_SERVICE);
                msg_rsp->setRCode(RCode::RCODE_OK);
                conn->send(msg_rsp);
            }


            void topicCreate(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 构造主题对象, 添加映射关系管理;
                std::string topic_name = msg->topicKey();
                auto topic = std::make_shared<Topic>(topic_name);
                _topics.insert(std::make_pair(topic_name, topic));
            }

            void topicRemove(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 1.删除主题数据,  有哪些订阅者, 然后从订阅者中删除主题信息;
                // 2. 删除主题数据, 主题名称和主题对象的映射关系;
                std::string topic_name = msg->topicKey();
                Topic::ptr topic;
                std::vector<Subscriber::ptr> subscribers;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _topics.find(topic_name);
                    if (it == _topics.end())
                    {
                        return;
                    }
                    topic = it->second;
                    subscribers = it->second->subscribers;
                    _topics.erase(it);
                    for (auto &subscriber : subscribers)
                    {
                        subscriber->removeTopic(topic_name);
                    }
                }
            }

            bool topicSubscribe(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 先找到主题对象, 以及订阅对象;
                Topic::ptr topic;
                Subscriber::ptr subscribe;
                // 没有主题对象就报错, 没有找到订阅对象就是构造订阅对象;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(msg->topicKey());
                    if (topic_it == _topics.end())
                    {
                        return false;
                    }
                    topic = topic_it->second;
                    auto sub_it = _subscribes.find(conn);
                    if (sub_it != _subscribes.end())
                    {
                        subscribe = sub_it->second;
                    }
                    else
                    {
                        subscribe = std::make_shared<Subscriber>(conn);
                        _subscribes.insert(std::make_pair(conn, subscribe));
                    }
                }
                // 在主题对象中, 新增一个对象关联的连接, 在订阅者对象中新增一个订阅主题.
                topic->appendSubscribee(conn);
                subscribe->appendTopic(msg->topicKey());
                return true;
            }

            void topicCancel(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 先找到主题对象, 以及订阅对象;
                Topic::ptr topic;
                Subscriber::ptr subscribe;
                // 没有主题对象就报错, 没有找到订阅对象就是构造订阅对象;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(msg->topicKey());
                    if (topic_it != _topics.end())
                    {
                        topic = topic_it->second;
                    }
                    topic = topic_it->second;
                    auto sub_it = _subscribes.find(conn);
                    if (sub_it != _subscribes.end())
                    {
                        subscribe = sub_it->second;
                    }
                }
                // 从主题对象中删除当前订阅者连接, 订阅者信息中删除订阅者名称;
                if (topic)
                    topic->removeSubscriber(conn);
                if (subscribe)
                    subscribe->removeTopic(msg->topicKey());
            }

            bool topicPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                Topic::ptr topic;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(msg->topicKey());
                    if (topic_it == _topics.end())
                    {
                        return false;
                    }
                    topic = topic_it->second;
                }
                topic->pushMessage(msg);
                return true;
            }

        private:
            struct Subscriber
            {
                using ptr = std::shared_ptr<Subscriber>;
                std::mutex _mutex;
                BaseConnection::ptr _conn;
                std::unordered_set<std::string> _topics; // 订阅者所订阅的主题名称;

                Subscriber(const BaseConnection::ptr &c)
                    : _conn(c)
                {
                }
                void appendTopic(const std::string &topic_name)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _topics.insert(topic_name);
                }
                // 主题删除, 取消订阅的时候调用;
                void removeTopic(const std::string &topic_name)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _topics.erase(topic_name);
                }
            };
            struct Topic
            {
                using ptr = std::shared_ptr<Topic>;
                std::string _topic_name;
                std::mutex _mutex;
                std::unordered_set<Subscriber::ptr> _subscribes; // 当前主题的订阅者;
                Topic(const std::string &name) : _topic_name(name) {};
                std::string TopicName() { return _topic_name; };
                std::unordered_set<Subscriber::ptr> subscribers()
                {
                    return _subscribes;
                }
                void appendSubscribee(const Subscriber::ptr &subscribe)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _subscribes.insert(subscribe);
                }
                void removeSubscriber(const Subscriber::ptr &subscribe)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _subscribes.erase(subscribe);
                }
                void pushMessage(const BaseMessage::ptr &msg)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    for (auto &subscriber : _subscribes)
                    {
                        subscriber->_conn->send(msg);
                    }
                }
            };

        private:
            std::mutex _mutex;
            std::unordered_map<std::string, Topic::ptr> _topics;
            std::unordered_map<BaseConnection::ptr, Subscriber::ptr> _subscribes;
        };
    }
}